package com.sg.java.apps;

import com.sg.java.PropertiesUtil;
import com.sg.java.ResourcePath;
import com.sg.java.util.DateUtils;
import com.sg.java.util.HBaseUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.*;

public class GetDataFromHiveToHBase {

    private static final Logger log = LoggerFactory.getLogger(GetDataFromHiveToHBase.class);

    public static void main(String[] args) throws Exception {
        if (args != null && args.length != 0) {
            long total_s = System.currentTimeMillis();
            Properties prop = PropertiesUtil.createPropertiesFromResource("hive.jdbc.properties");
            log.info("prop:{}", prop);
            String url = prop.getProperty("url");
            String user = prop.getProperty("user");
            String password = prop.getProperty("password");
            org.apache.hadoop.hbase.client.Connection hBaseConn = HBaseUtils.getHBaseConn(PropertiesUtil.createPropertiesFromResource(ResourcePath.hbase_properties));
            Map<Integer, byte[]> point_HHmmss = DateUtils.pointHHmmssToMap();

            Class.forName("org.apache.hive.jdbc.HiveDriver");

            Arrays.stream(args).parallel().forEach(arg -> {
                try {
                    long table_s = System.currentTimeMillis();
                    String[] split = arg.split(":");
                    if (split.length != 4) {
                        throw new IllegalArgumentException("格式不正确，正确格式：hiveTableName:type:pageSize:dataDate1,dataDate2 ... std_uep_un15_02_amr_e_mp_vol_curve:U:100000:20220813,20220814 ...");
                    }
                    String tableName = split[0];
                    String type = split[1].toUpperCase();
                    int pageSize = Integer.parseInt(split[2]);
                    String[] dataDates = split[3].split(",");
                    if (dataDates.length == 0) {
                        throw new IllegalArgumentException("格式不正确，正确格式：table1Name:20220801,20220802 table2Name:20220801,20220802 ...");
                    }

                    Logger log = LoggerFactory.getLogger(tableName);
                    Connection hiveConn = DriverManager.getConnection(url, user, password);

                    String infix;
                    String uniqueField;
                    switch (type) {
                        case "U":
                            infix = "cms_volt_curve_";
                            uniqueField = "phase_flag";
                            break;
                        case "I":
                            infix = "cms_cur_curve_";
                            uniqueField = "phase_flag";
                            break;
                        case "P":
                            infix = "cms_power_curve_";
                            uniqueField = "data_type";
                            break;
                        default:
                            throw new RuntimeException("type必须是U或I或P");
                    }

                    //sql in 指定data_date
//                    StringJoiner sj = new StringJoiner(",");
                    //日期和表
                    Map<String, Table> dataDate_Table = new HashMap<>();
                    //日期和待插入数据
                    Map<String, List<Put>> dataDate_Puts = new HashMap<>();

                    for (String dataDate : dataDates) {
//                        sj.add("'" + dataDate + "'");
                        dataDate_Table.put(dataDate, hBaseConn.getTable(TableName.valueOf(HBaseUtils.withNamespace(infix + dataDate))));
                        dataDate_Puts.put(dataDate, new ArrayList<>(pageSize));
                    }

                    String lowerType = type.toLowerCase();

                    StringJoiner cols = new StringJoiner(",");
                    for (int i = 1; i <= 96; i++) {
                        cols.add(lowerType + i);
                    }

//                    final String sqlTemplate = "select %s,id,ds,org_no,%s from %s where ds in (%s) limit %d offset %d";
                    final String sqlTemplate = "select %s,id,ds,org_no,%s from %s where id > '%s' and ds = '%s' order by id asc limit %d";
                    int tempCount;
                    int offset = 0;
                    int round = 0;
                    long temp_s;
                    long query_s;
                    long query_e;
                    long iterator_s;
                    long iterator_e;
                    long insert_s;
                    long insert_e;
                    String sql;
                    PreparedStatement ps;
                    ResultSet rs;
                    byte[] info_column = Bytes.toBytes("info");
                    //U/I/P
                    byte[] type_column = Bytes.toBytes(type);
                    byte[] METER_ID_qualifier = Bytes.toBytes("METER_ID");
                    byte[] DATA_DATE_qualifier = Bytes.toBytes("DATA_DATE");
                    byte[] ORG_NO_qualifier = Bytes.toBytes("ORG_NO");
                    byte[] emptyValue = Bytes.toBytes("");
                    //PHASE_FLAG/DATA_TYPE
                    byte[] uniqueField_qualifier = Bytes.toBytes(uniqueField.toUpperCase());
                    String maxId = "0";
                    for (String ds : dataDates) {
                        List<Put> puts = dataDate_Puts.get(ds);
                        Table table = dataDate_Table.get(ds);
                        do {
                            temp_s = System.currentTimeMillis();
                            round++;
                            tempCount = 0;
//                        sql       = String.format(sqlTemplate, cols, uniqueField, tableName, sj, pageSize, offset);
                            sql = String.format(sqlTemplate, cols, uniqueField, tableName, maxId, ds, pageSize);
                            log.info("sql:{}", sql);
                            ps = hiveConn.prepareStatement(sql);
                            ps.setFetchSize(pageSize);
                            query_s    = System.currentTimeMillis();
                            rs         = ps.executeQuery();
                            query_e    = System.currentTimeMillis() - query_s;
                            iterator_s = System.currentTimeMillis();
                            long exe_s;
                            long exe_total = 0;
                            while (rs.next()) {
                                exe_s = System.currentTimeMillis();
                                tempCount++;
                                offset++;
                                //METER_ID
                                String id = rs.getString(97);
                                maxId = id;
                                //20220815
                                String dataDate = rs.getString(98);
                                String orgNo = rs.getString(99);
                                //1
                                String uniqueValue = rs.getString(100);
                                String row = id + "-" + dataDate + "-" + uniqueValue;
                                Put put = new Put(Bytes.toBytes(row));
                                put.addColumn(info_column, METER_ID_qualifier, id != null ? Bytes.toBytes(id) : emptyValue);
                                put.addColumn(info_column, DATA_DATE_qualifier, dataDate != null ? Bytes.toBytes(dataDate) : emptyValue);
                                put.addColumn(info_column, ORG_NO_qualifier, orgNo != null ? Bytes.toBytes(orgNo) : emptyValue);
                                //PHASE_FLAG or DATA_TYPE
                                put.addColumn(info_column, uniqueField_qualifier, uniqueValue != null ? Bytes.toBytes(uniqueValue) : emptyValue);
                                for (int i = 1; i <= 96; i++) {
                                    String value = rs.getString(i);
                                    if (StringUtils.isNotBlank(value)) {
                                        put.addColumn(type_column, point_HHmmss.get(i), Bytes.toBytes(value));
                                    }
                                }
                                dataDate_Puts.get(dataDate).add(put);
                                exe_total += System.currentTimeMillis() - exe_s;
                                if (tempCount >= pageSize) {
                                    break;
                                }
                            }
                            iterator_e = System.currentTimeMillis() - iterator_s;
                            insert_s   = System.currentTimeMillis();
                            try {
                                if (puts.size() != 0) {
                                    table.put(puts);
                                    Put samplePut = puts.get(0);
                                    log.info("dataDate:{}\thbase表:{}\t插入条数:{}\t样例数据:{}", ds, table.getName(), puts.size(), samplePut);
                                    puts.clear();
                                }
                            } catch (IOException e) {
                                throw new RuntimeException(e);
                            }
                            insert_e = System.currentTimeMillis() - insert_s;
                            log.info("此轮:{}\t此轮读写数:{}\t已读写总数:{}\t此轮总耗时:{}ms\t查询耗时:{}ms\t数据集迭代总耗时:{}ms\t迭代操作耗时:{}ms\t插入耗时:{}ms", round, tempCount, offset, System.currentTimeMillis() - temp_s, query_e, iterator_e, exe_total, insert_e);
                        } while (tempCount >= pageSize);
                    }
                    hiveConn.close();
                    log.info("hive表：{}数据迁移完成，耗时:{}ms", tableName, System.currentTimeMillis() - table_s);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            });
            hBaseConn.close();
            log.info("程序执行完毕，总耗时:{}ms", System.currentTimeMillis() - total_s);
        } else {
            throw new IllegalArgumentException("入参不能为空");
        }
    }

}
